package cn.linjianhui.demo.flink.example.Function;

import cn.linjianhui.demo.flink.example.model.ItemViewCount;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

public class TopNProcessFunction extends KeyedProcessFunction<Tuple, ItemViewCount, String> {
    private int topSize;

    private ListState<ItemViewCount> itemStates;

    private static final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");

    public TopNProcessFunction(int topSize) {
        this.topSize = topSize;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        // 状态的注册
        ListStateDescriptor<ItemViewCount> itemsStateDesc = new ListStateDescriptor<>(
                "itemState-state", ItemViewCount.class);
        itemStates =getRuntimeContext().getListState(itemsStateDesc);
    }

    @Override
    public void processElement(ItemViewCount itemViewCount, Context context, Collector<String> out) throws Exception {
        itemStates.add(itemViewCount);
//        System.out.println(String.format("%s:%d:%d", itemViewCount.getItemId(), itemViewCount.getWindowEnd(), itemViewCount.getWindowEnd() + 1));
        context.timerService().registerEventTimeTimer(itemViewCount.getWindowEnd() + 1);
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
//        System.out.println(String.format("OnTimer timestamp: %d", timestamp));
        List<ItemViewCount> allItems = new ArrayList<>();
        for(ItemViewCount item : itemStates.get()) {
            allItems.add(item);
        }
//        System.out.println(String.format("count: %d", allItems.size()));
        itemStates.clear();
        allItems.sort((a, b) -> (int)(b.getCount() - a.getCount()));
        String result = allItems.stream().limit(topSize).map(n ->
                String.format("count:%d, %d,%s,%d", allItems.size(), n.getItemId(), sdf.format(n.getWindowEnd()), n.getCount()))
                .collect(Collectors.joining("|"));
//        Thread.sleep(1000);
        out.collect(result);
    }
}
